package com.example.selector;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName SelectorThreadGroup
 * @Description TODO
 * @Author: XMK
 * @Date: 2020/9/28 9:26
 * @Version: 1.0
 */
public class SelectorThreadGroup {

    private SelectorThread[] selectors;
    private AtomicInteger atomicInteger = new AtomicInteger(0);

    private SelectorThreadGroup worker = this;

    public SelectorThreadGroup(int num) {
        selectors = new SelectorThread[num];

        for (int i = 0; i < num; i++) {
            selectors[i] = new SelectorThread(this);
            new Thread(selectors[i]).start();
        }
    }

    public void bind(int port) {

        try {
            ServerSocketChannel server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));

            System.out.println("绑定端口：" + port);

            register2SelectorV3(server);

        } catch (IOException e) {
            e.printStackTrace();
        }


    }

    public void register2Selector(Channel channel) {

        SelectorThread selector = next();
        try {
            selector.getLbq().put(channel);
            selector.getSelector().wakeup();

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private SelectorThread next() {
        return selectors[atomicInteger.incrementAndGet() % selectors.length];
    }

    public void register2SelectorV2(Channel channel) {
        try {
            if (channel instanceof ServerSocketChannel) {
                SelectorThread selector = selectors[selectors.length - 1];
                selector.getLbq().put(channel);
                selector.getSelector().wakeup();
            } else if (channel instanceof SocketChannel) {
                SelectorThread selector = nextV2();
                selector.getLbq().put(channel);
                selector.getSelector().wakeup();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private SelectorThread nextV2() {
        return selectors[atomicInteger.incrementAndGet() % (selectors.length - 1)];
    }

    public void register2SelectorV3(Channel channel) {

        try {
            if (channel instanceof ServerSocketChannel) {
                SelectorThread selector = next();
                selector.getLbq().put(channel);
                selector.getSelector().wakeup();
            } else if (channel instanceof SocketChannel) {
                SelectorThread selector = nextV3();
                selector.getLbq().put(channel);
                selector.getSelector().wakeup();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private SelectorThread nextV3() {
        int index = worker.atomicInteger.incrementAndGet() % worker.selectors.length;
        return worker.selectors[index];
    }

    public SelectorThreadGroup getWorker() {
        return worker;
    }

    public void setWorker(SelectorThreadGroup worker) {
        this.worker = worker;
    }
}
